package com.fxbin123.demo;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

/**
 * @author: fxbin
 * @datetime: 2018/6/22 17:10
 * @description:
 */
public class ProducerDemo {

    public static void main(String[] args) {

        producer();

    }

    public static void producer(){
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.232.99:9092,192.168.232.100:9092,192.168.232.101:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, Object> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 100; i++) {
            System.out.println(i+"...");
//            producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));

            /**
             *
             * 利用回调参数提供的请求完成时将调用的回调通知
             */
            ProducerRecord<String, Object> record = new ProducerRecord<String, Object>("my-topic", Integer.toString(i), i);
            producer.send(record,
                    new Callback() {
                        @Override
                        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                            if(e != null){
                                e.printStackTrace();
                                System.out.println("The offset of the record we just sent is: " + recordMetadata.offset());
                            }
                        }
                    }
            );
        }
        producer.close();
    }

}
